package com.monitor.zkm.core.listener;

import bean.LinuxInfo;
import bean.ProjectInfo;
import bean.VMMemory;
import com.alibaba.fastjson.JSON;
import com.monitor.zkm.dao.LinuxRepository;
import com.monitor.zkm.dao.ProjectRepository;
import com.monitor.zkm.dao.VMRepository;
import core.ParamsContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.net.InetSocketAddress;
import java.util.List;

@Slf4j
public class ZkmMessageListener implements MessageListenerOrderly {

    private ProjectRepository projectRepository;
    private VMRepository vmRepository;
    private LinuxRepository linuxRepository;
    private final int MAX_RECOVERTIMES = 3;

    public ZkmMessageListener(ParamsContext paramsContext){
        this.projectRepository = (ProjectRepository) paramsContext.getVal("projectRep");
        this.vmRepository = (VMRepository) paramsContext.getVal("vmRep");
        this.linuxRepository = (LinuxRepository) paramsContext.getVal("linuxRep");
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        for(MessageExt messageExt : list){
            InetSocketAddress sourceIP = (InetSocketAddress) messageExt.getBornHost();
            String relativeIp = messageExt.getProperty("relativeIp");
            try {
                switch (messageExt.getTags()){
                    case "vm":
                        VMMemory vmMemory = JSON.parseObject(messageExt.getBody(), VMMemory.class);
                        vmMemory.setSourceIp(sourceIP.getHostString());
                        vmMemory.setRelativeIp(relativeIp);
                        log.info("接受信息：{}",vmMemory.toJson());
                        vmRepository.save(vmMemory);
                        break;
                    case "linux":
                        LinuxInfo linuxInfo = JSON.parseObject(messageExt.getBody(),LinuxInfo.class);
                        linuxInfo.setSourceIp(sourceIP.getHostString());
                        linuxInfo.setRelativeIp(relativeIp);
                        log.info("接受信息：{}",linuxInfo.toJson());
                        linuxRepository.save(linuxInfo);
                        break;
                    case "project":
                        ProjectInfo projectInfo = JSON.parseObject(messageExt.getBody(),ProjectInfo.class);
                        projectInfo.setSourceIp(sourceIP.getHostString());
                        projectInfo.setRelativeIp(relativeIp);
                        log.info("接受信息：{}",projectInfo.toJson());
                        projectRepository.save(projectInfo);
                        break;
                    default:
                        log.info("不存在的信息！");
                }
            }catch (Exception e){
                e.printStackTrace();
                //如果超过重试次数则不再消费
                if(messageExt.getReconsumeTimes() == MAX_RECOVERTIMES){
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
